/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.processing.reduce.klsh.kmeaning.mappers;

import aiaudio.processing.base.helpers.Algorithmic;
import aiaudio.Application;
import aiaudio.database.tablegroup.InitialDataTableGroup;
import aiaudio.database.tables.DataMetricsTable;
import java.io.IOException;
import java.util.HashMap;
import java.util.Random;
import java.util.Vector;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;

/**
 *
 * @author nastya
 */
public class KMeansDictionaryMapper extends TableMapper<ImmutableBytesWritable, ImmutableBytesWritable> {

    public static final int DEFAULT_TEST_SET_SIZE = 100;
    public static final int DEFAULT_DICTIONARY_COUNT = 5;
    public static final int DEFAULT_DICTIONARY_SIZE = 10;
    public static final String PARAM_TEST_SET_SIZE = "testSetSize";
    public static final String PARAM_DICTIONARY_SIZE = "dictionarySize";
    public static final String PARAM_DICTIONARY_COUNT = "dictionaryCount";
    private long testSetSize;
    private int dictionaryCount;
    private int dictionarySize;
    private Vector<Long> indices = new Vector<Long>();
    private HashMap<Integer, Vector<Long>> dictionariesProjection;

    protected void projectValuesOnIndiciesVector(Vector<Integer> proj, Vector<Long> ind) {
        for (int l : proj) {
            ind.add(indices.get(l));
        }
    }

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        super.setup(context);
        testSetSize = context.getConfiguration().getLong(PARAM_TEST_SET_SIZE, DEFAULT_TEST_SET_SIZE);
        dictionaryCount = context.getConfiguration().getInt(PARAM_DICTIONARY_COUNT, DEFAULT_DICTIONARY_COUNT);
        dictionarySize = context.getConfiguration().getInt(PARAM_DICTIONARY_SIZE, DEFAULT_DICTIONARY_SIZE);
        collectTestIndices();
        collectDictionaryIndices();
    }

    private void collectTestIndices() throws IOException {
        HTable dataMetricsTable = Application.database().getByName(InitialDataTableGroup.getInstance().getDataMetricsTable()).getTable();
        long artistsCount = DataMetricsTable.getArtistsCount(dataMetricsTable);

        Random r = new Random();
        long nextIndex;
        while (indices.size() < testSetSize) {
            nextIndex = (long) (r.nextDouble() * artistsCount);
            if (!indices.contains(nextIndex)) {
                indices.add(nextIndex);
            }
        }
    }
    long currentIndex = 0;

    @Override
    protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
        for (Integer dictionaryIndex : dictionariesProjection.keySet()) {
            Vector<Long> dict = dictionariesProjection.get(dictionaryIndex);
            if (dict.contains(currentIndex)) {
                byte[] dictionaryKey = Bytes.toBytes(dictionaryIndex + "");
                context.write(new ImmutableBytesWritable(dictionaryKey), new ImmutableBytesWritable(key));
            }
        }
        currentIndex++;
    }

    private void collectDictionaryIndices() {
        dictionariesProjection = new HashMap<Integer, Vector<Long>>();
        Random r = new Random();
        for (int i = 0; i < dictionaryCount; i++) {
            Vector<Long> ind = new Vector<Long>();
            Vector<Integer> proj = Algorithmic.randomValues(dictionarySize, indices.size(), r);
            projectValuesOnIndiciesVector(proj, ind);
            dictionariesProjection.put(i, ind);
        }
    }
}
